/*

 * Licensed to the Apache Software Foundation (ASF) under one

 * or more contributor license agreements.  See the NOTICE file

 * distributed with this work for additional information

 * regarding copyright ownership.  The ASF licenses this file

 * to you under the Apache License, Version 2.0 (the

 * "License"); you may not use this file except in compliance

 * with the License.  You may obtain a copy of the License at

 *

 *     http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS,

 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

 * See the License for the specific language governing permissions and

 * limitations under the License.

 */

package com.bff.gaia.unified.sdk.extensions.sql.impl;



import com.bff.gaia.unified.sdk.extensions.sql.UnifiedSqlTable;

import com.bff.gaia.unified.sdk.extensions.sql.impl.rel.UnifiedIOSinkRel;

import com.bff.gaia.unified.sdk.extensions.sql.impl.rel.UnifiedIOSourceRel;

import com.bff.gaia.unified.sdk.extensions.sql.impl.utils.CalciteUtils;

import com.bff.gaia.unified.vendor.guava.com.google.common.collect.ImmutableMap;

import org.apache.calcite.adapter.java.AbstractQueryableTable;

import org.apache.calcite.linq4j.QueryProvider;

import org.apache.calcite.linq4j.Queryable;

import org.apache.calcite.plan.RelOptCluster;

import org.apache.calcite.plan.RelOptTable;

import org.apache.calcite.prepare.Prepare;

import org.apache.calcite.rel.RelNode;

import org.apache.calcite.rel.core.TableModify;

import org.apache.calcite.rel.type.RelDataType;

import org.apache.calcite.rel.type.RelDataTypeFactory;

import org.apache.calcite.rex.RexNode;

import org.apache.calcite.schema.ModifiableTable;

import org.apache.calcite.schema.SchemaPlus;

import org.apache.calcite.schema.TranslatableTable;



import java.util.Collection;

import java.util.List;

import java.util.Map;



/** Adapter from {@link UnifiedSqlTable} to a calcite Table. */

public class UnifiedCalciteTable extends AbstractQueryableTable

    implements ModifiableTable, TranslatableTable {

  private final UnifiedSqlTable unifiedTable;

  private final Map<String, String> pipelineOptions;



  UnifiedCalciteTable(UnifiedSqlTable unifiedTable, Map<String, String> pipelineOptions) {

    super(Object[].class);

    this.unifiedTable = unifiedTable;

    this.pipelineOptions = pipelineOptions;

  }



  public static UnifiedCalciteTable of(UnifiedSqlTable table) {

    return new UnifiedCalciteTable(table, ImmutableMap.of());

  }



  @Override

  public RelDataType getRowType(RelDataTypeFactory typeFactory) {

    return CalciteUtils.toCalciteRowType(this.unifiedTable.getSchema(), typeFactory);

  }



  @Override

  public RelNode toRel(RelOptTable.ToRelContext context, RelOptTable relOptTable) {

    return new UnifiedIOSourceRel(context.getCluster(), relOptTable, unifiedTable, pipelineOptions);

  }



  @Override

  public <T> Queryable<T> asQueryable(

	  QueryProvider queryProvider, SchemaPlus schema, String tableName) {

    throw new UnsupportedOperationException();

  }



  @Override

  public Collection getModifiableCollection() {

    return null;

  }



  @Override

  public TableModify toModificationRel(

      RelOptCluster cluster,

      RelOptTable table,

      Prepare.CatalogReader catalogReader,

      RelNode child,

      TableModify.Operation operation,

      List<String> updateColumnList,

      List<RexNode> sourceExpressionList,

      boolean flattened) {

    return new UnifiedIOSinkRel(

        cluster,

        table,

        catalogReader,

        child,

        operation,

        updateColumnList,

        sourceExpressionList,

        flattened,

        unifiedTable,

        pipelineOptions);

  }

}